/*
 * Click nbfs://nbhost/SystemFileSystem/Templates/Licenses/license-default.txt to change this license
 * Click nbfs://nbhost/SystemFileSystem/Templates/Classes/Class.java to edit this template
 */
package com.excesys.exsecs.handler;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.excesys.exsecs.access.constant.RedisConstant;
import com.excesys.exsecs.access.service.ProtocolTransitionService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author XuYangbo
 */
@Component
@Slf4j
@RequiredArgsConstructor
public class MqttCallbackHandle {

	private final ProtocolTransitionService protocolTransitionService;
	private final StringRedisTemplate redisTemplate;

	// 线程安全整数,用来标识consumer线程数
	private static AtomicInteger counter = new AtomicInteger(0);

	private final ExecutorService executorService = new ThreadPoolExecutor(5, 30, 1, TimeUnit.SECONDS,
			new ArrayBlockingQueue<>(1024), Executors.defaultThreadFactory(), new ThreadPoolExecutor.DiscardPolicy());


	public void handle(String topic, String payload) {
		//log.info("传入数据----{}", payload);
		if (StringUtils.isBlank(payload)) {
			return;
		}
		JSONObject payloadJson = JSONUtil.parseObj(payload);
		if (ObjectUtil.isEmpty(payloadJson.get(RedisConstant.UNIT_NO)) && ObjectUtil.isEmpty(payloadJson.get(RedisConstant.TIMESTAMP))){
			return;
		}
		if (topic.contains("/properties")){
			executorService.submit(new MessageConsumer(payloadJson, counter, protocolTransitionService));
		}

		if (topic.contains("/services")){
			log.info("服务调用后返回的数据----{}", payload);
			redisTemplate.opsForHash().put(RedisConstant.SERVICES_RESPONSE_LIST_KEY, payloadJson.get(RedisConstant.TRACE_ID), payload);
			// 设置过期时间30秒
			redisTemplate.expire(RedisConstant.SERVICES_RESPONSE_LIST_KEY, 30, TimeUnit.SECONDS);
		}
	}

}
